package com.youxiake.lock.zk;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CountDownLatch;

/**
 * @author jxbai
 * @date 2021/12/15
 */
public class ZooKeeperSession {

    /**
     * java多线程并发同步的一个工具类,会传递进去一些数字，比如说1,2 ，3 都可以
     * 然后await()，如果数字不是0，那么久卡住，等待
     * 其他的线程可以调用countDown()，减1
     * 如果数字减到0，那么之前所有在await的线程，都会逃出阻塞的状态.继续向下运行
     */
    private static final CountDownLatch CONNECT_SEMAPHORE = new CountDownLatch(1);

    private ZooKeeper zooKeeper;

    private ZooKeeperSession() {
        // 去连接zookeeper server，创建会话的时候，是异步去进行的
        // 所以要给一个监听器，说告诉我们什么时候才是真正完成了跟zk server的连接
        try {
            this.zooKeeper = new ZooKeeper("1.116.186.202:2181", 5000, watchedEvent -> {
                System.out.printf("Receive watched state: %s%n", watchedEvent.getState());
                if (Watcher.Event.KeeperState.SyncConnected == watchedEvent.getState()) {
                    CONNECT_SEMAPHORE.countDown();
                }
            });
            // 给一个状态CONNECTING，连接中
            System.out.printf("zookeeper state: %s%n", zooKeeper.getState());
            CONNECT_SEMAPHORE.await();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取分布式锁
     * @param key 分布式锁key
     */
    public void acquireDistributedLock(String key) {
        try {
            zooKeeper.create(key, "".getBytes(StandardCharsets.UTF_8),
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            System.out.printf("success to acquire lock , key: %s%n", key);
        } catch (Exception e) {
            // 如果已经加锁，会报错
            int count = 0;
            while (true) {
                try {
                    Thread.sleep(20);
                    zooKeeper.create(key, "".getBytes(StandardCharsets.UTF_8),
                            ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                } catch (Exception e2) {
                    count ++;
                    continue;
                }
                System.out.printf("success to acquire lock , key: %s, try times: %s%n", key, count);
                break;
            }
        }
    }

    /**
     * 释放分布式锁
     * @param key 分布式锁key
     */
    public void releaseDistributedLock(String key) {
        try {
            zooKeeper.delete(key, -1);
            System.out.println("release lock success");
        } catch (Exception e) {
            System.out.printf("release lock fail. %s%n", e.getMessage());
        }
    }

    /**
     * 初始化单例的便捷方法
     * @return ZooKeeperSession
     */
    public static ZooKeeperSession init() {
        return getInstance();
    }

    /**
     * 获取单例
     * @return ZooKeeperSession
     */
    private static ZooKeeperSession getInstance() {
        return Singleton.getInstance();
    }

    /**
     * 封装单例的静态内部类
     */
    private static class Singleton {

        private static final ZooKeeperSession INSTANCE;

        static {
            INSTANCE = new ZooKeeperSession();
        }

        public static ZooKeeperSession getInstance() {
            return INSTANCE;
        }
    }

    public static void main(String[] args) {
        ZooKeeperSession session = ZooKeeperSession.init();
        new Thread(() -> {
            session.acquireDistributedLock("/product-2");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            session.releaseDistributedLock("/product-2");
        }).start();

        new Thread(() -> {
            session.acquireDistributedLock("/product-2");
        }).start();
    }
}
